package derbyclnt.sockets;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.*;
import java.util.concurrent.atomic.*;

import static derbyclnt.Logger.*;

public class AsyncClientSocket {
	private String _host;
	private int _port;
	private SocketListener _listener;
	
	
	private Thread _connectingThread;
	private Thread _readingThread;
	private Thread _writingThread;
	
	AtomicBoolean _connectingStarted = new AtomicBoolean();
	AtomicBoolean _readingStarted = new AtomicBoolean();
	AtomicBoolean _writingStarted = new AtomicBoolean();
	
	private Socket _socket;
	private DataInputStream _input;
	private DataOutputStream _output;
	
	int _readingBytes = 0;
	byte[] _writingBytes = new byte[0];
	
	
	
	public AsyncClientSocket(String host, int port, SocketListener listener) {
		assert port > 1023;
		assert host != null && host.length() != 0;
		assert listener != null;
		
		this._host = host;
		this._port = port;
		this._listener = listener;
	}
	
	public boolean isValid ()
	{
		boolean valid = _socket != null && _readingStarted.get() && _writingStarted.get();
		assert(!valid || (valid && _input != null && _output != null));
		return valid;
	}
	
	public void connect() throws Exception{
		if (_socket != null)
			disconnect();
		
		assert(!_connectingStarted.get());
		assert(!_readingStarted.get());
		assert(!_writingStarted.get());
		
		assert(_connectingThread == null);
		assert(_readingThread == null);
		assert(_writingThread == null);
		
		assert(_socket == null);
		assert(_input == null);
		assert(_output == null);
		
		_connectingThread = new Thread(new ConnectingTask());
		_connectingThread.start();
	}
	
	public void disconnect()
	{
		if (_connectingThread != null) {
			_connectingThread.interrupt();
			try {
				_connectingThread.join();
			} catch (InterruptedException e) {
				loge("This exception is abnormal: " + e.getMessage());
			}
			_connectingThread = null;
		}
		if (_readingThread != null){
			_readingThread.interrupt();
			try {
				_readingThread.join();
			} catch (InterruptedException e1) {
				loge("This exception is abnormal: " + e1.getMessage());
			}
			_readingThread = null;
		}
		if (_writingThread != null) {
			_writingThread.interrupt();
			try {
				_writingThread.join();
			} catch (InterruptedException e1) {
				loge("This exception is abnormal: " + e1.getMessage());
			}
			_writingThread = null;
		}
		
		assert(_connectingStarted.get() == false);
		assert(_readingStarted.get() == false);
		assert(_writingStarted.get() == false);
		
		_connectingStarted.set(false);
		_readingStarted.set(false);
		_writingStarted.set(false);
		
		_readingBytes = 0;
		_writingBytes = new byte[0];
		
		if (_input != null) {
			try {
				_input.close();
			} catch (IOException e) {
				loge(e.getMessage());
			} finally {
				_input = null;
			}
		}
		if (_output != null){
			try {
				_output.close();
			} catch (IOException e) {
				loge(e.getMessage());
			} finally {
				_output = null;
			}
		}
		if (_socket != null) {
			try {
				_socket.close();
			} catch (IOException e) {
				loge(e.getMessage());
			} finally {
				_socket = null;
			}
		}
		
		_listener.onClosedStatus(true);
	}
	
	public void receive(int byteCount) throws Exception {
		synchronized (AsyncClientSocket.this) {
			assert(_input != null);
			assert(_output != null);
			assert(_readingBytes == 0);
			assert(_socket != null);
			assert(_readingStarted.get());
			assert(_readingThread != null && _readingThread.isAlive());
			
			if (_socket == null || !_socket.isConnected())
				throw new Exception ("Socket is disconnected. ");
			if (_readingBytes != 0)
				throw new Exception ("Previous data has not been received yet!! ");
			if (_readingStarted.get() != true)
				throw new Exception ("Reading thread is stopped.");
			if (byteCount <= 0)
				throw new Exception ("byteCount must be positive");
			
			_readingBytes = byteCount;
		}
	}
	
	public void send(byte[] data) throws Exception {
		synchronized (AsyncClientSocket.this) {
			assert(_input != null);
			assert(_output != null);
			assert(_writingBytes != null && _writingBytes.length == 0);
			assert(_socket != null);
			assert(_readingStarted.get());
			assert(_writingStarted.get());
			assert(_writingThread != null && _writingThread.isAlive());
			
			
			if (_socket == null || !_socket.isConnected())
				throw new Exception ("Socket is disconnected");
			if (_writingStarted.get() != true)
				throw new Exception ("Writing thread is stopped. ");
			if (_writingBytes.length > 0)
				throw new Exception ("Previous data has not been sent yet!!");
			if (data == null || data.length == 0)
				throw new Exception ("data must be nonempty. ");
			
			_writingBytes = data;
		}
	}
	
	private class ConnectingTask implements Runnable {
		@Override
		public void run() {
			logd("ConnectingTask.run started");
			assert _connectingStarted.get() == false;
			assert _input == null;
			assert _output == null;

			try {
				_connectingStarted.set(true);
				_socket = new Socket(_host, _port);
				_input = new DataInputStream(_socket.getInputStream());
				_output = new DataOutputStream(_socket.getOutputStream());
				
				_readingThread = new Thread(new ReadingTask());
				_writingThread = new Thread(new WritingTask());
				_readingThread.start();
				_writingThread.start();
				
				while (_readingStarted.get() == false && _writingStarted.get() == false); //TODO 
				_listener.onClosedStatus(false);
			} catch (Exception e) {
				loge(e.getMessage());
				disconnect();
			} finally {
				_connectingStarted.set(false);
			}
			logd("ConnectingTask.run finished");
		}
	}
	
	private class ReadingTask implements Runnable {
		@Override
		public void run() {
			logd("ReadingTask.run started");
			assert(_readingStarted.get() == false);
			try {
				_readingStarted.set(true);
				
				while(!Thread.currentThread().isInterrupted()){
					int readingBytesCopy = 0;
					synchronized(AsyncClientSocket.this){
						readingBytesCopy = _readingBytes;
					}
					assert(readingBytesCopy >= 0);
					if (readingBytesCopy == 0) {
						Thread.sleep(100);
					} else if (readingBytesCopy > _input.available()){
						Thread.sleep(100);
					} else {
						logd("some data was received: " + readingBytesCopy + " bytes");
						byte[] buffer = new byte[readingBytesCopy];
						int bytesRead = _input.read(buffer);
						
						synchronized (AsyncClientSocket.this) {
							assert(readingBytesCopy == _readingBytes);
							_readingBytes = 0;
						}
						if (bytesRead != readingBytesCopy)
							throw new Exception("Reading failed. ");
						
						assert(buffer.length == bytesRead);
						_listener.onDataReceived(buffer, true);
					}
				}
				
			} catch (Exception e) {
				loge(e.getMessage());
				_listener.onDataReceived(new byte[0], false);
			} finally {
				_readingStarted.set(false);
			}
			logd("ReadingTask.run finished");
		}
	}
	
	private class WritingTask implements Runnable {
		@Override
		public void run() {
			logd("WritingTask.run started");
			try {
				assert(_writingStarted.get() == false);
				_writingStarted.set(true);
				while (!Thread.currentThread().isInterrupted()) {
					int writingBytesCopy = 0;
					synchronized (AsyncClientSocket.this) {
						assert _writingBytes != null;
						writingBytesCopy = _writingBytes.length;
					}
					if (writingBytesCopy == 0) {
						Thread.sleep(100);
					} else  {
						synchronized (AsyncClientSocket.this) {
							_output.write(_writingBytes);
							_writingBytes = new byte[0];
						}
						_listener.onDataSent(true);
					}
				}
			} catch (Exception e) {
				loge(e.getMessage());
				_listener.onDataSent(false);
			} finally {
				_writingStarted.set(false);
			}
			logd("WritingTask.run finished");
		}
	}
}
